package com.jy.oms.framework.asyn.task.cluster.registry;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.RandomUtil;
import com.jy.oms.framework.asyn.task.cluster.config.ITaskConfigChangeListener;
import com.jy.oms.framework.asyn.task.cluster.config.ITaskConfigService;
import com.jy.oms.framework.asyn.task.cluster.model.RealWorkNode;
import com.jy.oms.framework.asyn.task.cluster.model.WorkNode;
import com.jy.oms.framework.asyn.task.cluster.model.WorkNodeEvent;
import com.jy.oms.framework.asyn.task.cluster.model.WorkNodeExecutableTask;
import com.jy.oms.framework.asyn.task.constants.CacheConstants;
import com.jy.oms.framework.asyn.task.constants.ConfigConstants;
import com.jy.oms.framework.asyn.task.core.ExecutorConfig;
import com.jy.oms.framework.asyn.task.core.ExecutorContext;
import com.jy.oms.framework.asyn.task.core.RedissonClientProxy;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerManager;
import com.jy.oms.framework.asyn.task.model.TaskGroup;
import com.jy.oms.framework.asyn.task.model.TaskHandlerWrapper;
import com.jy.oms.framework.thread.CloseableThread;
import com.jy.oms.framework.util.NetworkUtil;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RMap;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;
import java.util.stream.Collectors;

/**
 * 节点注册服务
 *
 * @author liuquan
 */
@Service
public class RegistryService implements IRegistryService, DisposableBean {

    @Autowired
    private ITaskConfigService configService;
    @Autowired
    private ITaskHandlerManager taskHandlerManager;

    private WorkNode workNode;

    @Override
    public void registryWorkNode() {
        //注册节点
        this.handleRedisRegistryNode();

        RealWorkNode realWorkNode = workNode.getRealWorkNode();
        List<WorkNodeExecutableTask> groupTasks = this.parseWorkNodeExecutableTask(true, realWorkNode.getNodeNum());
        List<WorkNodeExecutableTask> tasks = this.parseWorkNodeExecutableTask(false, realWorkNode.getNodeNum());
        workNode.setExecutableGroupTasks(groupTasks);
        workNode.setExecutableTasks(tasks);
        workNode.mergeWorkNodeExecutableTask();

        //监听配置变化
        this.listenConfigKeyChange(true, realWorkNode.getNodeNum());
        this.listenConfigKeyChange(false, realWorkNode.getNodeNum());

        //监听节点注册
        this.nodeRegistryListener(realWorkNode);

        Thread workNodeHeartbeatThread = new WorkNodeHeartbeatThread();
        workNodeHeartbeatThread.setName("workNodeHeartbeatThread");
        workNodeHeartbeatThread.setDaemon(true);
        workNodeHeartbeatThread.start();
    }

    private void nodeRegistryListener(RealWorkNode realWorkNode) {
        RTopic rTopic = this.getRTopic();
        //监听节点注册
        rTopic.addListener(WorkNodeEvent.class, (channel, event) -> {
            Integer nodeNum = event.getNodeNum();
            String nodeIp = event.getNodeIp();
            if (realWorkNode.getNodeNum().equals(nodeNum)) {
                return;
            }

            if (!event.isRegistry()) {
                workNode.getOtherClusterWorkNodes().remove(nodeNum);
                return;
            }

            List<WorkNodeExecutableTask> groupTasks = this.parseWorkNodeExecutableTask(true, nodeNum);
            List<WorkNodeExecutableTask> tasks = this.parseWorkNodeExecutableTask(false, nodeNum);
            RealWorkNode otherRealWorkNode = RealWorkNode.create().setNodeIp(nodeIp).setNodeNum(nodeNum);
            WorkNode otherWorkNode = WorkNode.create().setRealWorkNode(otherRealWorkNode);
            otherWorkNode.setExecutableGroupTasks(groupTasks);
            otherWorkNode.setExecutableTasks(tasks);
            otherWorkNode.mergeWorkNodeExecutableTask();
            workNode.getOtherClusterWorkNodes().put(nodeNum, otherWorkNode);

            this.listenConfigKeyChange(true, nodeNum);
            this.listenConfigKeyChange(false, nodeNum);
        });

        //发布节点注册事件
        rTopic.publish(new WorkNodeEvent(true, realWorkNode.getNodeIp(), realWorkNode.getNodeNum()));
    }

    private RTopic getRTopic() {
        RedissonClient redissonClient = RedissonClientProxy.getRedissonClient();
        RTopic rTopic = redissonClient.getTopic(CacheConstants.genCacheKey(CacheConstants.TASK_CONFIG_PUBLISHER_QUEUE));
        return rTopic;
    }

    private RMap<Integer, RealWorkNode> getRMap() {
        String cacheKey = CacheConstants.genCacheKey(CacheConstants.TASK_REGISTRY_NODE_UPDATOR_MAP);
        RedissonClient redissonClient = RedissonClientProxy.getRedissonClient();
        return redissonClient.getMap(cacheKey);
    }

    private void handleRedisRegistryNode() {
        String lockKey = CacheConstants.genCacheKey(CacheConstants.TASK_REGISTRY_NODE_UPDATOR_MAP + ":lock");
        RedissonClientProxy.lock(lockKey, () -> {
            //清理节点
            this.cleanRedisRegistryNode();
            //获取节点序号
            int nodeNum = this.getRedisRegistryNodeNum();
            RealWorkNode realWorkNode = RealWorkNode.create()
                    .setNodeIp(NetworkUtil.getLocalhostStr())
                    .setNodeNum(nodeNum)
                    .setRegistryTime(System.currentTimeMillis());
            workNode = WorkNode.create().setRealWorkNode(realWorkNode);

            this.registryRealNode(realWorkNode);

            return null;
        });
    }

    private void registryRealNode(RealWorkNode realWorkNode) {
        RMap<Integer, RealWorkNode> registryMap = this.getRMap();
        registryMap.put(realWorkNode.getNodeNum(), realWorkNode);
        registryMap.expire(5, TimeUnit.MINUTES);
    }

    private int getRedisRegistryNodeNum() {
        RMap<Integer, RealWorkNode> registryMap = this.getRMap();
        Map<Integer, RealWorkNode> nodeMap = registryMap.readAllMap();
        List<Integer> nodeNumList = nodeMap.entrySet().stream()
                .map(entry -> entry.getValue().getNodeNum())
                .collect(Collectors.toList());
        for (int i = 1; i < Integer.MAX_VALUE; i++) {
            if (!nodeNumList.contains(i)) {
                return i;
            }
        }

        return 1;
    }

    private void cleanRedisRegistryNode() {
        RMap<Integer, RealWorkNode> registryMap = this.getRMap();
        Map<Integer, RealWorkNode> nodeMap = registryMap.readAllMap();
        long currentTime = System.currentTimeMillis();
        Set<Integer> cleanKeys = new HashSet<>();
        for (Map.Entry<Integer, RealWorkNode> entry : nodeMap.entrySet()) {
            if (currentTime - entry.getValue().getRegistryTime() > 90_000) {
                cleanKeys.add(entry.getKey());
            }
        }

        for (Integer cleanKey : cleanKeys) {
            registryMap.remove(cleanKey);
        }
    }

    private List<WorkNodeExecutableTask> parseWorkNodeExecutableTask(boolean isGroup, int nodeNum) {
        String configKey = isGroup ? ConfigConstants.WORK_NODE_EXECUTABLE_TASK_GROUP_PREFIX : ConfigConstants.WORK_NODE_EXECUTABLE_TASK_PREFIX;
        String config = configService.get(configKey + nodeNum);
        if (StringUtils.isBlank(config)) {
            return null;
        }

        Predicate<String> predicate;
        if (isGroup) {
            Map<TaskGroup, List<TaskHandlerWrapper>> taskHandlerMap = taskHandlerManager.loadAllExecutableTaskHandler();
            Set<String> taskGroups = taskHandlerMap.keySet().stream().map(TaskGroup::getGroup).collect(Collectors.toSet());
            predicate = taskGroups::contains;
        } else {
            predicate = (e) -> taskHandlerManager.loadTaskHandler(e) != null;
        }

        List<String> tasks = Arrays.stream(config.split(","))
                .filter(StringUtils::isNotBlank)
                .map(StringUtils::trim)
                .filter(predicate)
                .distinct()
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(tasks)) {
            return null;
        }

        if (isGroup) {
            return tasks.stream().map(WorkNodeExecutableTask::new).collect(Collectors.toList());
        } else {
            Map<String, List<String>> groupTasks = new HashMap<>();
            for (String taskCode : tasks) {
                TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskCode);
                TaskGroup taskGroup = taskHandlerWrapper.getTaskGroup();
                String group = taskGroup.getGroup();
                if (groupTasks.get(group) == null) {
                    groupTasks.put(taskCode, new ArrayList<>());
                }
                groupTasks.get(group).add(taskCode);
            }
            return groupTasks.entrySet().stream()
                    .map(entry -> new WorkNodeExecutableTask(entry.getKey(), entry.getValue()))
                    .collect(Collectors.toList());
        }
    }

    private void listenConfigKeyChange(boolean isGroup, int nodeNum) {
        configService.addTaskConfigChangeListener(new ITaskConfigChangeListener() {
            @Override
            public String getConfigKey() {
                String configKey = isGroup ? ConfigConstants.WORK_NODE_EXECUTABLE_TASK_GROUP_PREFIX : ConfigConstants.WORK_NODE_EXECUTABLE_TASK_PREFIX;
                return configKey + nodeNum;
            }

            @Override
            public void onConfigChange(String configKey, String config) {
                List<WorkNodeExecutableTask> tasks = parseWorkNodeExecutableTask(isGroup, nodeNum);
                WorkNode updateWorkNode;
                if (workNode.getRealWorkNode().getNodeNum() == nodeNum) {
                    updateWorkNode = workNode;
                } else {
                    updateWorkNode = workNode.getOtherClusterWorkNodes().get(nodeNum);
                }

                if (updateWorkNode != null) {
                    if (isGroup) {
                        updateWorkNode.setExecutableGroupTasks(tasks);
                    } else {
                        updateWorkNode.setExecutableTasks(tasks);
                    }
                    updateWorkNode.mergeWorkNodeExecutableTask();
                }
            }
        });
    }

    @Override
    public void unRegistryWorkNode() {
        RMap<Integer, RealWorkNode> registryMap = this.getRMap();
        RealWorkNode realWorkNode = workNode.getRealWorkNode();
        registryMap.remove(realWorkNode.getNodeNum());

        //发布节点下线事件
        RTopic rTopic = this.getRTopic();
        rTopic.publish(new WorkNodeEvent(false, realWorkNode.getNodeIp(), realWorkNode.getNodeNum()));
    }

    @Override
    public void destroy() throws Exception {
        ExecutorConfig executorConfig = ExecutorContext.getExecutorConfig();
        if (executorConfig != null && executorConfig.isEnableCluster()) {
            unRegistryWorkNode();
        }
    }

    @Override
    public WorkNode getWorkNode() {
        return workNode;
    }

    private class WorkNodeHeartbeatThread extends CloseableThread {
        @Override
        public void runCloseable() {
            RealWorkNode realWorkNode = workNode.getRealWorkNode();
            realWorkNode.setRegistryTime(System.currentTimeMillis());
            registryRealNode(realWorkNode);

            ThreadUtil.sleep(RandomUtil.randomInt(50, 60), TimeUnit.SECONDS);
        }
    }

}
